/*
 * Middleware
 * Copyright (C) 2010/11
 * Riccardo Pulvirenti, Giuseppe Ravidà, Andrea Tino
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANDABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see http://www.gnu.org/licenses
 */

/*!
 * File: worker.cpp
 */

#include <iostream>
#include <string>
#include <exception>
#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/archive/binary_iarchive.hpp>
#include <boost/archive/binary_oarchive.hpp>
#include <cstdlib>

#include "utils.hpp"
#include "worker.hpp"
#include "discovery_proxy_factory.hpp"

using namespace middleware;

//-----------------------------------------------------------------------------
// Constructors and destructor
/* Default constructor */
Worker::Worker(WorkerDescriptor descriptor) {
	this->_worker_descriptor = descriptor;
	this->_config.worker_port_todisc = "127.0.0.1";
	this->_config.worker_port_todisc = "0";
}
/* Parameter constructor */
Worker::Worker(WorkerDescriptor descriptor, WorkerConfig conf) {
	this->_worker_descriptor = descriptor;
	this->_config.worker_ip_todisc = conf.worker_ip_todisc;
	this->_config.worker_port_todisc = conf.worker_port_todisc;
}
/* Destructor */
Worker::~Worker() {

}
//-----------------------------------------------------------------------------
// Private members
/* Gets a random port (orphan) */
std::string Worker::createPort() {
	return utils::GetRandomIntegralNumberStr(kWorkerMinManPort,
			kWorkerMaxManPort);
}
/* Thread execution flow */
void Worker::exec() {
	// Execution flow for Worker
	std::cout << kWorkerEtq << "Running!" << std::endl;
	this->bonjour(); /* Initialization sequence */
	try {
		/* Release thread for listening to incoming requests and enqueueing
		 * workflows in the queue. */
		std::cout << kWorkerEtq << "Releasing Enqueuer..." << std::endl;
		this->_enqueuer_thread = boost::shared_ptr<boost::thread>(
				new boost::thread(boost::bind(&Worker::exec_enqueuer, this)));
		// Cycling and dequeueing from queue
		for (;;) {
			std::cout << kWorkerEtq << "Dequeueing loop, " <<
					"Dequeueing task..." << std::endl;
			TaskDescriptor td = this->_task_synqueue.dequeue(); /* Blocking */
			std::cout << kWorkerEtq << "Dequeued task id=" << td.get_id() <<
					". TQueue has now " << this->_task_synqueue.size() <<
					" tasks." << " Now managing dequeued task..." << std::endl;
			// A thread will manage the task (multithread)
			boost::shared_ptr<boost::thread>(new boost::thread(
					boost::bind(&Worker::ManageTask, this, td)));
			//this->ManageTask(td); /* Managing this flow (singlethread) */
		}
	} catch(std::exception& e) {
		std::cout << kWorkerEtq << "Error: " << e.what() << std::endl;
		std::cout << kWorkerEtq << "Now exiting!" << std::endl;
	}
}
/* Thread execution flow for enqueuer | Sharing a SynQueue */
void Worker::exec_enqueuer() {
	using boost::asio::ip::tcp;
	std::cout << kWorkerEtq << kWorker1Etq << "Running!" << std::endl;
	try {
		/* Listen on set port for incoming connections and retrieve data to
		 * put in the queue */
		int port = (int)(this->_worker_descriptor.get_address().get_port());
		boost::asio::io_service io_s;
		tcp::acceptor td_acceptor(io_s, tcp::endpoint(tcp::v4(), port));
		// Looping for accepting incoming requests
		for (;;) {
			tcp::iostream td_stream;
			TaskDescriptor td;
			std::cout << kWorkerEtq << kWorker1Etq <<
					"Waiting for task to be sent..." << std::endl;
			td_acceptor.accept(*(td_stream.rdbuf()));
			boost::archive::binary_iarchive ia(td_stream);
			ia >> td; /* Here I have the task */
			td_stream.close();
			std::cout << kWorkerEtq << kWorker1Etq <<
					"Task descriptor arrived: TID=" << td.get_id() <<
					". Now enqueueing..."  << std::endl;
			// Now enqueueing the workflow
			this->_task_synqueue.enqueue(td);
			std::cout << kWorkerEtq << kWorker1Etq <<
					"Task descriptor " << td.get_id() << " enqueued!"
					<< std::endl;
		}
	} catch(std::exception& e) {
		std::cout << kWorkerEtq << kWorker1Etq <<
				"Error: " << e.what() << std::endl;
		std::cout << kWorkerEtq << kWorker1Etq <<
				"Program now dying..." << std::endl;
		exit(1);
	}
}
/* Sending periodically worker status to discovery */
void Worker::exec_workerstatus_manager() {
	using boost::asio::ip::tcp;
	/* Activate for a certain period of time and, when self-triggered, check
	 * for some changes to have happened in worker status, if so, collect
	 * info and send them to discovery */
	try {
		std::cout << kWorkerEtq << kWorker2Etq << "Running!" << std::endl;
		bool init = true; /* First time the control loop is executed? */
		WorkerDescriptor prev_wdesc; /* To check diffs with previous status */
		// Creating factory and getting proxy
		std::cout << kWorkerEtq << kWorker2Etq <<
				"Configuring proxy factory..." << std::endl;
		DiscoveryProxyFactory proxy_factory(this->_config.worker_ip_todisc,
				this->_config.worker_port_todisc);
		std::cout << kWorkerEtq << kWorker2Etq <<
				"Proxy factory configured: Ip=" <<
				this->_config.worker_ip_todisc << " and Port=" <<
				this->_config.worker_port_todisc << "!" << std::endl;
		std::cout << kWorkerEtq << kWorker2Etq <<
				"Getting proxy from factory..." << std::endl;
		WorkerDiscoveryProxy* proxy = proxy_factory.getProxy();
		std::cout << kWorkerEtq << kWorker2Etq << "Got proxy!" << std::endl;
		std::cout << kWorkerEtq << kWorker2Etq <<
				"Entering worker status manager loop..." << std::endl;
		for (;;) {
			std::cout << kWorkerEtq << kWorker2Etq << "Reactivated..." <<
					std::endl;
			if (init) {
				/* If this is the first time the control loop is executed,
				 * then we must send this worker's status because it is a
				 * registration to discovery and not a status refresh */
				init = false;
				std::cout << kWorkerEtq << kWorker2Etq <<
							"Worker needs to register to Discovery..." <<
							std::endl;
				proxy->update_worker(this->_worker_descriptor);
				prev_wdesc = this->_worker_descriptor;
			} else {
				/* Normal refresh message to be sent */
				// First of all, tell me whether status changed.
				if (this->_worker_descriptor != prev_wdesc) {
					// Changes detected ==> Refresh
					std::cout << kWorkerEtq << kWorker2Etq <<
							"Changes detected!" <<
							" Sending wdesc to Discovery..." <<	std::endl;
					proxy->update_worker(this->_worker_descriptor); /* Send */
					prev_wdesc = this->_worker_descriptor; /* Update */
					std::cout << kWorkerEtq << kWorker2Etq <<
							"Worker Descriptor on Discovery refreshed!" <<
							std::endl;
				} else {
					// No changes
					std::cout << kWorkerEtq << kWorker2Etq <<
							"No Changes detected!" <<
							" Going to sleep and reactivate later..." <<
							std::endl;
				}
			}
			// Waiting and reactivating then again
			boost::this_thread::sleep(boost::posix_time::milliseconds(
					kWorkerStatusManagerRefreshPeriodMsec));
		}
	} catch(std::exception& e) {
		/* Proxy reference is automatically destroys as the factory gets
		 * out of scope */
		std::cout << kWorkerEtq << kWorker2Etq << "Error: " << e.what() <<
				std::endl;
		std::cout << kWorkerEtq << kWorker2Etq << "Now exiting!" << std::endl;
	}
}
/* Managing a task (can be athread) */
void Worker::ManageTask(TaskDescriptor task) {
	std::cout << kWorkerEtq << "TManager: " << "Managing task "
			<< task.get_id() << std::endl;
}
/* Bonjour */
void Worker::bonjour() {
	std::cout << kWorkerEtq << "Initialization sequence now started..." <<
				std::endl;
	std::cout << kWorkerEtq << "Configuration loaded: bind_tomanIP: this=" <<
			this->_worker_descriptor.get_address().get_ip()  <<
			", bind_tomanPORT: " <<
			this->_worker_descriptor.get_address().get_port() <<
			"; todiscIP: " << this->_config.worker_ip_todisc <<
			", todiscPORT: " <<	this->_config.worker_port_todisc << std::endl;
	std::cout << kWorkerEtq << "Initialization completed!" << std::endl;
	std::cout << kWorkerEtq << "Collecting capabilities..." << std::endl;
	std::cout << kWorkerEtq << "Worker capabilities: " <<
			"CPU: " <<
			this->_worker_descriptor._capabilities.capab_cpu_mhz <<
			" MHz; " <<	"RAM: " <<
			this->_worker_descriptor._capabilities.capab_ram_mb << " MB; " <<
			"Cores: " <<
			this->_worker_descriptor._capabilities.capab_core_num <<
			" cores; " << "Processors: " <<
			this->_worker_descriptor._capabilities.capab_cpu_num <<
			" cpus; " << "Distance: " <<
			this->_worker_descriptor._capabilities.capab_hops <<
			" hops; " << "Bandwidth: " <<
			this->_worker_descriptor._capabilities.capab_bandwidth <<
			" Kbit/s." << std::endl;
	std::cout << kWorkerEtq << "Report ended!" << std::endl;
}
